package org.yuanzheng.tableAndSql

import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.sources.CsvTableSource
import org.apache.flink.types.Row

/*纯SQL*/
object TestSql {
  def main(args: Array[String]): Unit = {
    //创建使用flink原生
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)

    //隐式转换
    import org.apache.flink.streaming.api.scala._

    //读取数据
    val tableSource = new CsvTableSource("E:\\Java_Study\\@m-code\\ScalaFlink\\src\\main\\resources\\station.log",
      Array[String]("sid", "callOut", "callIn", "callType", "callTime", "duration"),
      Array(Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.LONG, Types.LONG))

    //使用存粹的SQL
    tableEnv.registerTableSource("t_station_log", tableSource) //注册表
    val result: Table = tableEnv.sqlQuery("select sid,sum(duration) as dSun from t_station_log where callType='success' group by sid")
    tableEnv.toRetractStream[Row](result).filter(_._1 == true).print()
    tableEnv.execute("dSum")
  }
}
